package org.flink.time;

import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.flink.api.Item;
import org.flink.api.MyStreamSource;

import java.util.concurrent.TimeUnit;

/**
 * @Author: TongRui乀
 * @Date: 2021/1/23 15:43
 * @description： 测试watermark
 */
public class WaterMarkDemo {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置使用事件事件
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        env.socketTextStream("192.168.199.151", 8888)
                .map(item->{
                    String[] split = item.split(",");
                    Integer id = Integer.valueOf(split[0]);
                    String name = split[1];
                    Long timestamp = Long.valueOf(split[2]);
                    return new Item(name, id, timestamp);
                })
                // 设置watermark  参数是最大乱序事件
                .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Item>(Time.seconds(3)) {
                    @Override
                    public long extractTimestamp(Item item) {
                        return item.getTimestamp();
                    }
                })
        .keyBy(Item::getName)
        // 设置时间窗口
        .timeWindow(Time.seconds(5))
        .reduce((i1,i2)->{
            Item item = new Item();
            item.setId(i1.getId() + i2.getId());
            item.setName(i1.getName());
            item.setTimestamp(System.currentTimeMillis());
            item.setDesc("sum per 5 seconds current name is " + item.getName());
            return item;
        })
        .printToErr();

        env.execute();

    }
}
